package cn.flink.exercise;

import org.apache.flink.table.api.*;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * 页面浏览数统计
 * 
 * @author mayong
 * @date 2023/05/07
 **/
public class pageVewFlinkStat {

    public static void main(String[] args) {

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().build();

        TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings);

        Schema schema = Schema.newBuilder()
                .column("ip", DataTypes.STRING())
                .column("userId",DataTypes.STRING())
                .column("eventTime",DataTypes.STRING())
                .column("method",DataTypes.STRING())
                .column("url",DataTypes.STRING())
                .build();

        tableEnvironment.createTemporaryTable("sourceTable", TableDescriptor.forConnector("kafka")
                .schema(schema)
                .option("topic", "access_log")
                .option("properties.bootstrap.servers","bigdata01:9092,bigdata02:9092,bigdata03:9092")
                .option("properties.group.id", "log")
                .option("scan.startup.mode", "earliest-offset")
                .option("format", "json")
                .build()
        );

        Table table = tableEnvironment.from("sourceTable").select($("url"), $("eventTime"));

        Table tableResult = table.window(Slide.over(lit(10).minute()).every(lit(5).second()).on($("eventTime")).as("w"))
                .groupBy($("url"), $("w"))
                .select($("url"), $("url").count());

        tableResult.execute().print();


    }

}
